package org.diy.traffic.sinks;

import java.sql.BatchUpdateException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.logging.Level;

import org.diy.traffic.updates.TrafficUpdate;
import org.diy.traffic.updates.TrafficUpdate.UnknownDataException;
import org.diy.traffic.util.TLog;
public class DiyFluSink extends TrafficSink {

	private static String sinkType = "DIYFLU_DB_SINK";
	
	private static String mysqlDbName;
	private static String mysqlUsername;
	private static String mysqlPassword;

	private static HashMap<Integer, ArrayList<String>> blacklistTerms = new HashMap<Integer,ArrayList<String>>();
	
	private static Connection dbConn = null;
	
	public DiyFluSink(int rateLimit) throws TrafficSinkException {
		super(sinkType, rateLimit);
		
		initDB();
	}

	public void initDB() throws TrafficSinkException {
		try {
			dbConn = DriverManager.getConnection(
					String.format("jdbc:mysql://localhost/%s?user=%s&password=%s", mysqlDbName, mysqlUsername, mysqlPassword)
					);
		} catch (SQLException e) {
			TLog.log(Level.WARNING, e);
			throw new TrafficSinkException(String.format("Can't connect to the database with dbname: '%s', username: '%s', password: '%s'",
					mysqlDbName, mysqlUsername, mysqlPassword));

		}
	}
	
	public static DiyFluSink createFromConfigData(HashMap<String, String> configParams) throws TrafficSinkException {
		mysqlDbName   = configParams.get("db");
		mysqlUsername = configParams.get("username");
		mysqlPassword = configParams.get("password");
		
		if ((mysqlDbName == null) || (mysqlUsername == null) || (mysqlPassword == null)) {
			TLog.log(Level.SEVERE, "DiyFluSink requires a 'db', 'username', and 'password' parameter for configuration! None provided.");
			return null;
		}
		
		return new DiyFluSink(5000);
	}
	
	@Override
	public boolean processUpdate(TrafficUpdate update)
			throws TrafficSinkException {

		
		
		Statement stmt = null;
		ResultSet rs   = null;
		try {
			
			//System.out.println(String.format("****************Got a Flu Update: %s, %s, %s, %s", 
			//		update.getData("status_id"), update.getData("city_id"), update.getData("text"), update.getData("mentioner")));
		
			
			stmt = dbConn.createStatement();
			
			if (update.getMessage().equals("TYPE_MENTION")) {
				
				if (!containsBlacklistTerm(update)) {
				
					int query_result = stmt.executeUpdate(
							String.format("INSERT INTO mentions (mentioned_at, message, link, exact_location, phrase_id, query_id, city_id, mentioner, created_at, updated_at, source_id) "+
									"VALUES (\"%s\", \"%s\", \"%s\", \"%s, %s\", %d, %d, %d, \"%s\", %s, %s, %s) ON DUPLICATE KEY UPDATE updated_at=updated_at",
									(String)update.getData("mentioned_at"),
									scrubStatusMessage((String) update.getData("text")),
									(String)update.getData("link"),
									(String)update.getData("city_name"),
									(String)update.getData("country_name"),
									Integer.parseInt((String)update.getData("phrase_id")),
									Integer.parseInt((String)update.getData("query_id")),
									Integer.parseInt((String)update.getData("city_id")),
									(String)update.getData("mentioner"),
									"NOW()",
									"NOW()",
									Integer.parseInt((String) update.getData("status_id")))
							);
				}
			} else if (update.getMessage().equals("TYPE_META")) {
				ArrayList<TrafficUpdate> update_list = (ArrayList<TrafficUpdate>) update.getData("update_list");
				batchInsertUpdates(update_list);
			} else {
				
				TLog.log(Level.FINER, String.format("Updating record for city id %d on date %s with count %s", update.getData("city_id"), update.getData("date"), update.getData("count")));
				
				/**
				stmt.executeUpdate(
						String.format("INSERT INTO hourly_counts (city_id, date, last_found_id, count) VALUES (%s, \"%s\", %s, %s) ON DUPLICATE KEY UPDATE count=count+%d",
								update.getData("city_id"), update.getData("date"), update.getData("last_found_id"), update.getData("count"), update.getData("count") )
								);
				**/
				
				
				rs = stmt.executeQuery(
						String.format("SELECT * FROM hourly_counts WHERE city_id=%s AND date=\"%s\"",
								update.getData("city_id"),
								update.getData("date"))
						);
				rs.last();
				if(rs.getRow() <= 0) {
					// no entry exists for this date for this city
					TLog.log(Level.FINEST, String.format("Creating new record for %s in city id %s with count of %s ",
											update.getData("date"),update.getData("city_id"),update.getData("count"))
							);
					
					stmt.executeUpdate(
							String.format("INSERT INTO hourly_counts (city_id, date, last_found_id, count) VALUES (%s, \"%s\", %s, %s)",
									update.getData("city_id"), update.getData("date"), update.getData("last_found_id"), update.getData("count"))
									);
				} else {
					TLog.log(Level.FINEST, String.format("Updating existing record for %s in city id %s with %s additional counts and last_id %s",
							update.getData("date"),update.getData("city_id"),update.getData("count"), update.getData("last_found_id"))
							);
					rs.first();
					int current_count = rs.getInt("count");
					stmt.executeUpdate(
							String.format("UPDATE hourly_counts SET count=%s,last_found_id=%s WHERE city_id=%s AND date=\"%s\"",
									String.valueOf(current_count + (Integer)update.getData("count")),
									update.getData("last_found_id"),
									update.getData("city_id"),
									update.getData("date"))
							);
				
				}
				
			}
			//TLog.log(Level.FINE, String.format("DiyFluSink modified %d rows in the database", query_result));
			//System.out.println("Data successfully entered into db!!! :) :) :)");
		} catch (UnknownDataException e) {
			// TODO Auto-generated catch block
			TLog.log(Level.WARNING, e);
			throw new TrafficSinkException("DiyFluSource: Error while parsing data from update!");
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			TLog.log(Level.WARNING, e);
			throw new TrafficSinkException("DiyFluSource: Error entering update into database!");
		} finally {
			try {
				if (stmt != null) stmt.close();
				if (rs != null)   rs.close();
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				TLog.log(Level.WARNING, e);
			}
		}
		
		update = null;
		System.gc();
		return true;
	}
	
	private String scrubStatusMessage(String statusMessage) {
		statusMessage = statusMessage.replace("\\", "\\\\");
		return statusMessage;
	}

	private boolean containsBlacklistTerm(TrafficUpdate update) {
		
		try {
			String update_text = (String) update.getData("text");
			
			Integer phrase_id = Integer.valueOf((String)update.getData("phrase_id"));
			
			
			ArrayList<String> blacklist_for_phrase = new ArrayList<String>();
			if (blacklistTerms.containsKey(phrase_id)) 
					blacklist_for_phrase.addAll(blacklistTerms.get(phrase_id));
			if (blacklistTerms.containsKey(Integer.valueOf(-1))) 
				blacklist_for_phrase.addAll(blacklistTerms.get(Integer.valueOf(-1)));
			
			
			Iterator<String> i = blacklist_for_phrase.iterator();
			while (i.hasNext()) {
				String blacklist_term = i.next();
				if (update_text.toLowerCase().indexOf(String.format("%s", blacklist_term.toLowerCase())) >= 0) { 
					TLog.log(Level.FINEST, 
							String.format("***Scrapping tweet '%s' because it contans blacklist term '%s'", update_text, blacklist_term)
							);
					return true;
				}
			}
		
		} catch (UnknownDataException e) {
			TLog.log(Level.WARNING, "Error checking update against blacklist, can't find phrase id!");
			return false;
		}
		
		return false;
	}
	
	public int batchInsertUpdates(ArrayList<TrafficUpdate> updates){

		Iterator<TrafficUpdate> update_iterator = updates.iterator();
		


	    //conn.setAutoCommit(false);
		
	    String query = "INSERT INTO all_tweets (mentioned_at, message, link, exact_location, city_id, mentioner, created_at, updated_at, source_id) "+
				"VALUES (?, ?, ?, ?, ?, ?, NOW(), NOW(), ?) ON DUPLICATE KEY UPDATE updated_at=updated_at";
				
	    	
	    //"insert into test(name, desg) values(?, ?)";
        // conn.setAutoCommit(false);

	    PreparedStatement ps = null;
	    try {
	    	ps = dbConn.prepareStatement(query);
	    } catch (SQLException sqle) {
	    	TLog.log(Level.WARNING, "SQL exception while creating prepared statemetn!");
			TLog.log(Level.WARNING, sqle.getMessage());
			return 0;
	    }
	    	
		while (update_iterator.hasNext()) {
			TrafficUpdate t_update = update_iterator.next();
			
			try {
				ps.setString(1, String.format("%s", (String)t_update.getData("mentioned_at")));
				ps.setString(2, String.format("%s", scrubStatusMessage((String) t_update.getData("text"))));
				ps.setString(3, String.format("%s", (String)t_update.getData("link")));
				ps.setString(4, String.format("%s, %s", (String)t_update.getData("city_name"),(String)t_update.getData("country_name")));
				ps.setInt(5, Integer.parseInt((String)t_update.getData("city_id")));
				ps.setString(6, String.format("%s", (String)t_update.getData("mentioner")));
				ps.setInt(7, Integer.parseInt((String) t_update.getData("status_id")));
				
				ps.addBatch();
				
			} catch (BatchUpdateException be) {
				TLog.log(Level.WARNING, "Batch update exception while doing batch tweet insert.");
				TLog.log(Level.WARNING, be.getMessage());
				continue;
			} catch (SQLException sqle) {
				TLog.log(Level.WARNING, "SQL exception while doing batch tweet insert.");
				TLog.log(Level.WARNING, sqle.getMessage());
				continue;
			} 	catch (UnknownDataException ude) {
				TLog.log(Level.WARNING, "Unkown data while doing batch tweet insert.");
				TLog.log(Level.WARNING, ude.getMessage());
				continue;
			} 
			
			

		
			/**
			int query_result = stmt.executeUpdate(
					String.format("INSERT INTO all_tweets (mentioned_at, message, link, exact_location, city_id, mentioner, created_at, updated_at, source_id) "+
							"VALUES (\"%s\", \"%s\", \"%s\", \"%s, %s\", %d, \"%s\", %s, %s, %s) ON DUPLICATE KEY UPDATE updated_at=updated_at",
							(String)t_update.getData("mentioned_at"),
							scrubStatusMessage((String) t_update.getData("text")),
							(String)t_update.getData("link"),
							(String)t_update.getData("city_name"),
							(String)t_update.getData("country_name"),
							Integer.parseInt((String)t_update.getData("city_id")),
							(String)t_update.getData("mentioner"),
							"NOW()",
							"NOW()",
							Integer.parseInt((String) t_update.getData("status_id")))
					);
			**/
		}
		
	    int[] updateCounts=null;
		try {
			updateCounts = ps.executeBatch();
		} catch (SQLException sqle) {
			TLog.log(Level.WARNING, "SQL exception while  executing batch.");
			TLog.log(Level.WARNING, sqle.getMessage());
		} finally {
			try { if (ps!=null) ps.close(); }
			catch (SQLException e) {}
		}
		
		if (updateCounts!=null) {
			System.err.println("Length:" + updateCounts.length);
		    return updateCounts.length;
		} else return 0;
	    //conn.commit();
	    

	}
	
	public static void refreshBlackListTerms() {
		HashMap<Integer, ArrayList<String>> backupMap = new HashMap<Integer, ArrayList<String>>(blacklistTerms);
		
		blacklistTerms.clear();
		Statement blacklist_stmt = null;
		ResultSet blacklist_rs   = null;
		
		try {
			blacklist_stmt = dbConn.createStatement();
			blacklist_rs = blacklist_stmt.executeQuery("SELECT * FROM blacklist");
			
			while (blacklist_rs.next()) {
				int phrase_id = blacklist_rs.getInt("phrase_id");
				Integer phrase_id_integer = Integer.valueOf(phrase_id);
				if (!blacklistTerms.containsKey(phrase_id_integer)) {
					blacklistTerms.put(phrase_id_integer, new ArrayList<String>());
				}
				blacklistTerms.get(phrase_id_integer).add(blacklist_rs.getString("term"));
				
			}
			
		} catch (SQLException e) {
			blacklistTerms = backupMap;
			TLog.log(Level.SEVERE, "Error refreshing list of blacklist terms!!");
			TLog.log(Level.SEVERE, e);
		} finally {
			try {
				if (blacklist_rs != null) blacklist_rs.close();
				if (blacklist_stmt != null) blacklist_stmt.close();
			} catch (SQLException sqle) {}
		}
		
		/**
		Iterator bi = blacklistTerms.keySet().iterator();
		while (bi.hasNext()) {
			Integer i = (Integer) bi.next();
			ArrayList<String> terms = blacklistTerms.get(i);
			Iterator termi = terms.iterator();
			while (termi.hasNext()) {
				System.out.println(i + ": " + termi.next());
			}

		}
		**/
		
	}

}
